package kalp.web;

import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.twitter.TwitterReceiver;

import scala.collection.Seq;
import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.auth.Authorization;

public class ExtendedTwReciever extends TwitterReceiver {

	private static final long serialVersionUID = 1L;
	private Authorization auth;

	public ExtendedTwReciever(Authorization twitterAuth, Seq<String> filters,
			StorageLevel storageLevel) {
		super(twitterAuth, filters, storageLevel);
		auth = twitterAuth;
	}

	@Override
	public void onStart() {
		TwitterStream ts = new TwitterStreamFactory().getInstance(auth);
		ts.addListener(new StatusListener() {

			public void onException(Exception ex) {
				restart("...");
			}

			public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
			}

			public void onStatus(Status status) {
				store(status);
			}

			public void onStallWarning(StallWarning warning) {
			}

			public void onScrubGeo(long userId, long upToStatusId) {
			}

			public void onDeletionNotice(
					StatusDeletionNotice statusDeletionNotice) {
			}
		});

		FilterQuery fq = new FilterQuery();
		// fq.locations(new double[][] { { -180, -90 }, { 180, 90 } });
		// ts.filter(fq);
		String[] keywordTotrack = {"windows"};
		fq.track(keywordTotrack);
//		ts.filter(fq);
		ts.sample();
	}

}
